package com.grf.kafka.java.serializer;

import com.grf.kafka.java.KafkaConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @author grf
 * @apiNote 用户饭序列化
 * @since 2024/10/25
 */
@Slf4j
public class UserConsumer {
    public static void main(String[] args) {
        // PART1:设置发送者相关属性
        Properties props = new Properties();
        // 此处配置的kafka的端口
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstant.BOOTSTRAP_SERVERS);
        // 每个消费者要指定一个group
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        // 配置key的序列化类
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"com.grf.kafka.java.serializer.UserDeserializer");
        Consumer<String, User> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(KafkaConstant.TOPIC));
        while (true){
            ConsumerRecords<String, User> records = consumer.poll(Duration.ofNanos(100));

            // PART3: 处理消息
            for(ConsumerRecord<String, User> record : records){
                log.info("offset is {};key is {};value is {}",record.offset(),record.key(),record.value());
            }
            // 提交offset，消息就不会重复推送
            consumer.commitSync();//同步提交，表示必须等到offset提交后再
        }
    }
}
